package org.apache.flink.client.cluster;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import javax.annotation.Nonnull;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.ConsumingStatus;
import org.apache.flink.api.common.ResourceStatus;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.allocation.ResourceAllocationStatistics;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.minicluster.MiniCluster;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Client to interact with a {@link MiniCluster}. */
public class MiniClusterClient implements ClusterClient<MiniClusterClient.MiniClusterId> {

    private static final Logger LOG = LoggerFactory.getLogger(MiniClusterClient.class);

    private final MiniCluster miniCluster;
    private final Configuration configuration;

    public MiniClusterClient(
            @Nonnull Configuration configuration, @Nonnull MiniCluster miniCluster) {
        this.configuration = configuration;
        this.miniCluster = miniCluster;
    }

    @Override
    public Configuration getFlinkConfiguration() {
        return new Configuration(configuration);
    }
    
    
    @Override
	public CompletableFuture<ResourceStatus> getResourceStatus() {
		return miniCluster.getResourceStatus();
	}

	@Override
	public CompletableFuture<ResourceAllocationStatistics> getResourceAllocationStatistics() {
        return miniCluster.requestResourceAllocationStatistics();
	}

	@Override
    public CompletableFuture<Acknowledge> cancel(ConsumerID jobId) {
        return miniCluster.cancel();
    }

    @Override
    public CompletableFuture<ConsumingStatus> getConsumingStatus(ConsumerID jobId) {
        return CompletableFuture.completedFuture(ConsumingStatus.RUNNING);
    }

    @Override
    public void close() {}

    @Override
    public MiniClusterClient.MiniClusterId getClusterId() {
        return MiniClusterId.INSTANCE;
    }

    @Override
    public void shutDownCluster() {
        try {
            miniCluster.closeAsync().get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            LOG.error("Error while shutting down cluster", e);
        }
    }

    /** The type of the Cluster ID for the local {@link MiniCluster}. */
    public enum MiniClusterId {
        INSTANCE
    }

	@Override
	public String getWebInterfaceURL() {
		return null;
	}
}
